package zisu.netty.practice.RPC.client;

import lombok.extern.slf4j.Slf4j;
import zisu.netty.practice.RPC.client.codec.RpcRequest;
import zisu.netty.practice.RPC.client.codec.RpcResponse;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.locks.AbstractQueuedSynchronizer;
import java.util.concurrent.locks.ReentrantLock;

/**
 * @Desc: ""
 * @Author: caixiang
 * @DATE: 2021/1/21 10:29
 */
@Slf4j
public class RpcFuture implements Future<Object> {

    private RpcRequest rpcRequest;

    private RpcResponse rpcResponse;

    //就是拿到结果后 可以通知给 那些需要知道这些结果 的地方（这些地方实现这些回调接口 ，供这里调用。）
    private List<RpcCallBack> pendingCallBacks = new ArrayList<>();

    //自定义锁（这把锁 是 锁 get() 方法 和 done() 方法）
    private Sync sync;

    //也是锁（这把锁 是锁 invokeCallBacks()  和  addCallBacks() 这两个方法的，这两个方法不能同步进行。）
    private ReentrantLock lock = new ReentrantLock();

    //整体请求事件阈值
    private static final long TIME_THRESHOLD = 5000;

    //执行runCallbacks 任务的线程池
    private ThreadPoolExecutor executor = new ThreadPoolExecutor(16,16,60,TimeUnit.SECONDS,new ArrayBlockingQueue<Runnable>(65536));

    private long startTime;
    private long endTime;

    public RpcFuture(RpcRequest request){
        this.rpcRequest = request;
        this.startTime = System.currentTimeMillis();
        this.sync = new Sync();
    }

    //实际的回调处理 (done 方法和 get方法应该是一个互斥的状态)
    public void done(RpcResponse response) {
        this.rpcResponse = response;
        //独占式释放同步状态，该方法会在释放同步状态之后，将同步队列中第一个节点包含的线程唤醒
        boolean release = sync.release(1);
        if(release){
            invokeCallbacks();
        }
        //整体rpc调用的耗时时间
        long costTime = System.currentTimeMillis() - startTime;
        if(costTime> TIME_THRESHOLD){
            log.warn("the rpc response time is to slow,requestId:"+this.rpcRequest.getRequestId());
        }
    }

    //依次执行回调函数处理
    private void invokeCallbacks() {
        lock.lock();
        try {
            for(RpcCallBack callBack :pendingCallBacks){
                runCallBack(callBack);
            }
        }finally {
            lock.unlock();
        }

    }

    private void runCallBack(RpcCallBack callBack) {
        //为了不改变原始数据，，另起备份。
        final RpcResponse rpcResponse = this.rpcResponse;
        executor.submit(new Runnable() {
            @Override
            public void run() {
                if(rpcResponse.getThrowable() != null){
                    callBack.success(rpcResponse.getResult());
                }else {
                    callBack.failure(rpcResponse.getThrowable());
                }
            }
        });
    }

    @Override
    public boolean cancel(boolean mayInterruptIfRunning) {
        throw new UnsupportedOperationException();
    }

    @Override
    public boolean isCancelled() {
        throw new UnsupportedOperationException();
    }

    @Override
    public boolean isDone() {
        return sync.isDone();
    }

    //https://blog.csdn.net/qq_38293564/article/details/80480688
    @Override
    public Object get() throws InterruptedException, ExecutionException {
        //会 一直等待，直到拿到许可
        //独占式获取同步状态，如果当前线程获取同步状态成功，则由该方法返回，否则，将会进入同步队列等待，该方法将会调用可重写的tryAcquire(int arg)方法
        sync.acquire(-1);
        if(this.rpcResponse != null){
            return this.rpcResponse.getResult();
        }else {
            return null;
        }
    }

    @Override
    public Object get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
        boolean success = sync.tryAcquireNanos(-1, unit.toNanos(timeout));
        if(success){
            if(this.rpcResponse != null){
                return this.rpcResponse.getResult();
            }else {
                return null;
            }
        }else {
            throw new RuntimeException("timeout exception requestId:"+this.rpcRequest.getRequestId()+ ",className:"+this.rpcRequest.getClassName()+",methodName:"+this.rpcRequest.getMethodName());
        }
    }


    /**
     * 可以在应用过程中 添加回调处理函数。
     * @param rpcCallBack
     * @return
     */
    public RpcFuture addCallBack(RpcCallBack rpcCallBack){
        lock.lock();
        try {
            if(isDone()){
                runCallBack(rpcCallBack);
            }else {
                this.pendingCallBacks.add(rpcCallBack);
            }
        }finally {
            lock.unlock();
        }

        return this;
    }

    //自定义一把锁
    class Sync extends AbstractQueuedSynchronizer{

        private static final long serialVersionUID = 7354653937265687794L;

        private final int done = 1;
        private final int pending = 0;

        //许可证明
        @Override
        protected boolean tryAcquire(int acquires){
            //如果状态就是done  那么说明 结束了，可以拿到许可证了，否者返回false
            return getState() == done ? true:false;
        }

        //释放许可
        @Override
        protected boolean tryRelease(int release){
            if(getState() == pending){
                //比较当前状态（AbstractQueuedSynchronizer里的）,如果当前这个状态是等于pending状态的，那么把这个状态update为done状态
                //如果更新成功 就返回true，如果当前状态部位pending状态 就返回false
                if(compareAndSetState(pending,done)){
                    return true;
                }
            }
            return false;
        }

        public boolean isDone(){
            return getState() == done;
        }


    }
}
